#include <unistd.h>
#include <mosquitto.h>
#include "service.h"
#include "z_logger.h"
#include "aliot.h"
#include "aiot_devinfo_api.h"
#include "aiot_dm_api.h"
#include "sqlite.h"
#include "sysinfo.h"
#include "dm_private.h"
#include "cJSON.h"
#include "z_memory.h"
#include "gpio.h"

static void dm_recv_generic_reply(void *dm_handle, const aiot_dm_recv_t *recv, void *userdata) {
    log_trace("msg_id = %ld, code = %d, data = %.*s, message = %.*s",
              recv->data.generic_reply.msg_id,
              recv->data.generic_reply.code,
              recv->data.generic_reply.data_len,
              recv->data.generic_reply.data,
              recv->data.generic_reply.message_len,
              recv->data.generic_reply.message);
}

static int _gpio_fd = -1;

static void _modbus_set_rts(modbus_t *ctx, int on) {
    if (_gpio_fd <= 0) return;

    gpio_set_value(_gpio_fd, on);
    usleep(500);
}

typedef struct metric {
    char *metric;
    uint16_t value;
} metric_t;

metric_t *metric_new() {
    return z_calloc(1, sizeof(metric_t));
}

//modbus写入寄存器部分，控制操作
static int modbus_write(modbus_t *modbus, device_t *device, metric_t *metric) {
    int rc = 0;
    for (int i = 0; i < sizeof(device->_product->thing_model) / sizeof(device->_product->thing_model[0]); ++i) {
        vector_t *tag_list = &device->_product->thing_model[i];
        if (!vector_size(tag_list)) {
            continue;
        }
        for (int y = 0; y < vector_size(tag_list); ++y) {
            thing_model_t *tag = vector_get(tag_list, y);
            //根据产品id和metric唯一确定一列thing_model表中数据,tag->offset为寄存器地址
            if (device->_product->super.id != tag->product || metric->metric != tag->metric) {
                continue;
            }
            rc = modbus_write_registers(modbus, tag->offset, 1, &metric->value);
            if (rc < 0) {
                log_error("modbus写入出错：modbus_write_registers = %d\n", rc);
                return rc;
            }
        }
    }
    return rc;
}

//根据seqiot整合控制参数，完成控制命令
static int modbus_control(channel_t *channel, device_t *device, metric_t *metric) {
    _gpio_fd = gpio_open(channel->serial.gpio, "direction");
    modbus_t *modbus = modbus_new_rtu(channel->serial.device, channel->serial.baud,
                                      (1 == channel->serial.parity) ? 'O' :
                                      (2 == channel->serial.parity ? 'E' : 'N'),
                                      channel->serial.data_bits, channel->serial.stop_bits);
    modbus_rtu_set_serial_mode(modbus, MODBUS_RTU_RS485);
    modbus_rtu_set_rts(modbus, MODBUS_RTU_RTS_UP);
    modbus_rtu_set_custom_rts(modbus, _modbus_set_rts);
    modbus_rtu_set_rts_delay(modbus, 500);
    modbus_set_error_recovery(modbus, MODBUS_ERROR_RECOVERY_LINK | MODBUS_ERROR_RECOVERY_PROTOCOL);
    int rc = modbus_connect(modbus);
    if (rc < 0) {
        log_error("modbus_connect = %d\n", rc);
        return -1;
    }
    rc = modbus_set_slave(modbus, device->slave_id);
    if (rc < 0) {
        log_error("modbus_set_slave = %d\n", rc);
        return -1;
    }
    rc = modbus_set_response_timeout(modbus, 3, 0);
    if (rc < 0) {
        log_error("modbus_set_response_timeout = %d\n", rc);
        return -1;
    }
    rc = modbus_write(modbus, device, metric);
    close(_gpio_fd);
    gpio_unexport(channel->serial.gpio);
    modbus_close(modbus);
    modbus_free(modbus);
    return rc;
}

//拼接控制json字符串
static int assemble_json(cJSON *json, channel_t *channel, device_t *device, metric_t *metric) {
    for (int i = 0; i < sizeof(device->_product->thing_model) / sizeof(device->_product->thing_model[0]); ++i) {
        vector_t *tag_list = &device->_product->thing_model[i];
        if (vector_size(tag_list)) {
            for (int y = 0; y < vector_size(tag_list); ++y) {
                thing_model_t *tag = vector_get(tag_list, y);
                //根据产品id和metric唯一确定一列thing_model表中数据,tag->offset为寄存器地址
                if (device->_product->super.id == tag->product && metric->metric == tag->metric) {
                    cJSON_AddNumberToObject(json, "value", metric->value);
                    cJSON_AddNumberToObject(json, "channel_id", channel->super.id);
                    cJSON_AddNumberToObject(json, "offset", tag->offset);
                    cJSON_AddNumberToObject(json, "timestamp", time(NULL));
                    return 1;
                }
            }
        }
    }
    return 0;
}

static int mosquitto_sub(channel_t *channel, device_t *device, metric_t *metric) {
    mosquitto_lib_init();
    struct mosquitto *mosq = mosquitto_new("dev_control", false, NULL);
    mosquitto_connect(mosq, "127.0.0.1", 1883, 60);
    while (running) {
        int rc = mosquitto_loop(mosq, -1, 10);
        if (MOSQ_ERR_NO_CONN == rc || MOSQ_ERR_CONN_LOST == rc) {
            mosquitto_reconnect(mosq);
        }
        cJSON *cjson = cJSON_CreateObject();
        //拼接json字符串
        assemble_json(cjson, channel, device, metric);
        char *payload = cJSON_PrintUnformatted(cjson);
        rc = mosquitto_publish(mosq, NULL, "/user/control", sizeof(payload), payload, 1/*QoS*/, false);
        mosquitto_disconnect(mosq);
        mosquitto_destroy(mosq);
        mosquitto_lib_cleanup();
        return rc;
    }
    return -1;
}

static int recv_control(cJSON *json, const aiot_dm_recv_t *recv, seqiot_t *seqiot) {
    for (int i = 0; i < cJSON_GetArraySize(json); ++i) {
        cJSON *item = cJSON_GetArrayItem(json, i);
        metric_t *metric = metric_new();
        metric->metric = item->string;
        if (cJSON_Number == item->type)
            metric->value = item->valueint;
        for (int k = 0; k < vector_size(&seqiot->channels); ++k) {
            channel_t *channel = vector_get(&seqiot->channels, k);
            for (int j = 0; j < vector_size(&channel->devices); ++j) {
                device_t *device = vector_get(&channel->devices, j);
                //寻找对应device列
                if (recv->device_name == device->device_key && device->channel == channel->super.id) {
//                    return modbus_control(channel, device, metric);
                    return mosquitto_sub(channel, device, metric);
                }
            }
        }
    }
    return -1;
}

static void dm_recv_property_set(void *dm_handle, const aiot_dm_recv_t *recv, void *userdata) {
    log_trace("msg_id = %ld, params = %.*s",
              recv->data.property_set.msg_id,
              recv->data.property_set.params_len,
              recv->data.property_set.params);

    aiot_dm_msg_t msg;
    cJSON *json = cJSON_Parse(recv->data.property_set.params);
    seqiot_t *seqiot = userdata;
    int rc = recv_control(json, recv, seqiot);
    if (rc < 0) {
        log_error("rc=%d", rc);
    }

    bzero(&msg, sizeof(aiot_dm_msg_t));
    msg.type = AIOT_DMMSG_PROPERTY_SET_REPLY;
    msg.data.property_set_reply.msg_id = recv->data.property_set.msg_id;
    msg.data.property_set_reply.code = 200;
    msg.data.property_set_reply.data = "{}";
    int32_t res = aiot_dm_send(dm_handle, &msg);
    aliot_error_check("aiot_dm_send", res);
}

static void dm_recv_async_service_invoke(void *dm_handle, const aiot_dm_recv_t *recv, void *userdata) {
    log_trace("msg_id = %ld, service_id = %s, params = %.*s",
              recv->data.async_service_invoke.msg_id,
              recv->data.async_service_invoke.service_id,
              recv->data.async_service_invoke.params_len,
              recv->data.async_service_invoke.params);

    /* TODO:
        *
        * 注意: 如果用户在回调函数外进行应答, 需要自行保存msg_id, 因为回调函数入参在退出回调函数后将被SDK销毁, 不可以再访问到
        */
    aiot_dm_msg_t msg;

    bzero(&msg, sizeof(aiot_dm_msg_t));
    msg.type = AIOT_DMMSG_ASYNC_SERVICE_REPLY;
    msg.data.async_service_reply.msg_id = recv->data.async_service_invoke.msg_id;
    msg.data.async_service_reply.code = 200;
    msg.data.async_service_reply.service_id = "TEST";
    msg.data.async_service_reply.data = "{\"dataA\": 20}";
    int32_t res = aiot_dm_send(dm_handle, &msg);
    aliot_error_check("aiot_dm_send", res);
}

static void dm_recv_sync_service_invoke(void *dm_handle, const aiot_dm_recv_t *recv, void *userdata) {
    log_trace("msg_id = %ld, rrpc_id = %s, service_id = %s, params = %.*s",
              (unsigned long) recv->data.sync_service_invoke.msg_id,
              recv->data.sync_service_invoke.rrpc_id,
              recv->data.sync_service_invoke.service_id,
              recv->data.sync_service_invoke.params_len,
              recv->data.sync_service_invoke.params);

    /* TODO:
        *
        * 注意: 如果用户在回调函数外进行应答, 需要自行保存msg_id和rrpc_id字符串, 因为回调函数入参在退出回调函数后将被SDK销毁, 不可以再访问到
        */
    aiot_dm_msg_t msg;

    bzero(&msg, sizeof(aiot_dm_msg_t));
    msg.type = AIOT_DMMSG_SYNC_SERVICE_REPLY;
    msg.data.sync_service_reply.rrpc_id = recv->data.sync_service_invoke.rrpc_id;
    msg.data.sync_service_reply.msg_id = recv->data.sync_service_invoke.msg_id;
    msg.data.sync_service_reply.code = 200;
    msg.data.sync_service_reply.service_id = "SetLightSwitchTimer";
    msg.data.sync_service_reply.data = "{}";
    int32_t res = aiot_dm_send(dm_handle, &msg);
    aliot_error_check("aiot_dm_send", res);
}

static void dm_recv_raw_data(void *dm_handle, const aiot_dm_recv_t *recv, void *userdata) {
    log_trace("raw data len = %d", recv->data.raw_data.data_len);
    /* TODO: 以下代码演示如何发送二进制格式数据, 若使用需要有相应的数据透传脚本部署在云端 */
    /*
    {
        aiot_dm_msg_t msg;
        uint8_t raw_data[] = {0x01, 0x02};

        bzero(&msg, sizeof(aiot_dm_msg_t));
        msg.type = AIOT_DMMSG_RAW_DATA;
        msg.data.raw_data.data = raw_data;
        msg.data.raw_data.data_len = sizeof(raw_data);
        aiot_dm_send(dm_handle, &msg);
    }
    */
}

static void dm_recv_raw_sync_service_invoke(void *dm_handle, const aiot_dm_recv_t *recv, void *userdata) {
    log_trace("raw sync service rrpc_id = %s, data_len = %d",
              recv->data.raw_service_invoke.rrpc_id,
              recv->data.raw_service_invoke.data_len);
}

static void dm_recv_raw_data_reply(void *dm_handle, const aiot_dm_recv_t *recv, void *userdata) {
    log_trace("receive reply for up_raw msg, data len = %d",
              recv->data.raw_data.data_len);
    /* TODO: 用户处理下行的二进制数据, 位于recv->data.raw_data.data中 */
}

/* 用户数据接收处理回调函数 */
static void dm_recv_handler(void *dm_handle, const aiot_dm_recv_t *recv, void *userdata) {
    log_trace("type = %d", recv->type);

    switch (recv->type) {
        case AIOT_DMRECV_GENERIC_REPLY:/* 属性上报, 事件上报, 获取期望属性值或者删除期望属性值的应答 */
            dm_recv_generic_reply(dm_handle, recv, userdata);
            break;
        case AIOT_DMRECV_PROPERTY_SET: /* 属性设置 */
            dm_recv_property_set(dm_handle, recv, userdata);
            break;
        case AIOT_DMRECV_ASYNC_SERVICE_INVOKE:/* 异步服务调用 */
            dm_recv_async_service_invoke(dm_handle, recv, userdata);
            break;
        case AIOT_DMRECV_SYNC_SERVICE_INVOKE: /* 同步服务调用 */
            dm_recv_sync_service_invoke(dm_handle, recv, userdata);
            break;
        case AIOT_DMRECV_RAW_DATA: /* 下行二进制数据 */
            dm_recv_raw_data(dm_handle, recv, userdata);
            break;
        case AIOT_DMRECV_RAW_SYNC_SERVICE_INVOKE:/* 二进制格式的同步服务调用, 比单纯的二进制数据消息多了个rrpc_id */
            dm_recv_raw_sync_service_invoke(dm_handle, recv, userdata);
            break;
        case AIOT_DMRECV_RAW_DATA_REPLY: /* 上行二进制数据后, 云端的回复报文 */
            dm_recv_raw_data_reply(dm_handle, recv, userdata);
            break;
        default:
            break;
    }
}

/* 属性上报函数 */
int32_t dm_send_property_post(void *dm_handle, char *params) {
    aiot_dm_msg_t msg;

    bzero(&msg, sizeof(aiot_dm_msg_t));
    msg.type = AIOT_DMMSG_PROPERTY_POST;
    msg.data.property_post.params = params;

    return aiot_dm_send(dm_handle, &msg);
}

/* 事件上报函数 */
int32_t dm_send_event_post(void *dm_handle, char *event_id, char *params) {
    aiot_dm_msg_t msg;

    bzero(&msg, sizeof(aiot_dm_msg_t));
    msg.type = AIOT_DMMSG_EVENT_POST;
    msg.data.event_post.event_id = event_id;
    msg.data.event_post.params = params;

    return aiot_dm_send(dm_handle, &msg);
}

int32_t dm_send_property_batch_post(void *dm_handle, char *params) {
    aiot_dm_msg_t msg;

    bzero(&msg, sizeof(aiot_dm_msg_t));
    msg.type = AIOT_DMMSG_PROPERTY_BATCH_POST;
    msg.data.property_post.params = params;

    return aiot_dm_send(dm_handle, &msg);
}

static void *_dm_service_init(aliot_service_args_t *args) {
    void *dm_handle = aiot_dm_init();

    aiot_dm_setopt(dm_handle, AIOT_DMOPT_MQTT_HANDLE, args->mqtt_handle);
    aiot_dm_setopt(dm_handle, AIOT_DMOPT_RECV_HANDLER, dm_recv_handler);
    aiot_dm_setopt(dm_handle, AIOT_DMOPT_USERDATA, args->seqiot);

    return dm_handle;
}

static void _dm_service_fini(void **handle) {
    aiot_dm_deinit(handle);
}

static void *_dm_service_start(void *handle) {
    dm_handle_t *dm_handle = handle;
    seqiot_t *seqiot = dm_handle->userdata;
    vector_t *info_list = &seqiot->sysinfos;

    while (running) {
        time_t cur_time = time(NULL);
        if (0 == cur_time % 60) {
            cJSON *cjson = cJSON_CreateObject();
            for (int i = 0; i < vector_size(info_list); ++i) {
                double value;
                sys_info_t *info = vector_get(info_list, i);
                if (sys_info_exec_double(info->cmd, &value) != 0) {
                    log_error("id=%s, cmd=[%s]", info->id, info->cmd);
                    continue;
                }
                //比较当前值与记录值是否相等，改变则上传
                if (value == info->_value) {
                    continue;
                }
                log_debug("id=%s, value=%f", info->id, value);
                info->_value = value;
                vector_set(info_list, i, info);
                cJSON_AddNumberToObject(cjson, info->id, value);
            }
            if (NULL != cjson->child && NULL != cjson->child->string) {
                char *json = cJSON_PrintUnformatted(cjson);
                int32_t res = dm_send_property_post(handle, json);
                aliot_error_check("dm_send_property_post", res);
                cJSON_free(json);
            }
            cJSON_Delete(cjson);
        }
    }
    return NULL;
}

static aliot_service_t _dm_service = {
        .name = DATA_MODEL_MODULE_NAME,
        .init = _dm_service_init,
        .fini = _dm_service_fini,
        .start = _dm_service_start,
        .stop = NULL,
};

aliot_service_t *dm_service_get() {
    return &_dm_service;
}

